package com.ytktt.study.week12.task4.kafka.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.messaging.Message;

/**
 * 默认kafka消费异常处理器
 *
 * @author ytk
 */
@Slf4j
public class DefaultKafkaConsumerErrorHandler implements ConsumerAwareListenerErrorHandler {

    @Override
    public Object handleError(Message<?> message, ListenerExecutionFailedException exception,
                              Consumer<?, ?> consumer) {
        log.error("consumer error: {}", exception.getLocalizedMessage());
        // 有分区信息、偏移信息、Order信息等
        log.warn("save payload {} in db", message.getPayload());
        // 目前无用 没有开启@SendTo
        return message;
    }
}
